package com.taobao.analyse.reduce;

import java.io.IOException;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import com.taobao.analyse.constants.SplitConstants;

public class ItemPriceChangedReducer extends MapReduceBase implements
		Reducer<LongWritable, Text, LongWritable, Text> {
	
	@Override
	public void reduce(LongWritable key, Iterator<Text> values,
			OutputCollector<LongWritable, Text> output, Reporter reporter)
			throws IOException {
		Map<String,Integer> dateToUpdateCount = new HashMap<String,Integer>();
		Map<String,Integer> dateToUpdatePriceCount = new HashMap<String,Integer>();
		StringBuilder sb = new StringBuilder();
		//用于统计以rate开关的数据，即统计编辑次数
		StringBuilder sb_rate = new StringBuilder();
		
		StringBuilder sb_date = new StringBuilder();
		
//		if(values.hasNext()){
//			output.collect(key,new Text(values.next().toString()));
//			return;
//		}
		
		while(values.hasNext()){
			String valueText = values.next().toString();
			if(valueText.startsWith("rate")){
				recordUpdateCount(valueText,dateToUpdateCount,dateToUpdatePriceCount);
			}else{
				
				//截取时间点
				addDateTimeMark(sb_date,valueText);
				
				//detail详情暂时不记录输出
				//detail@1338607656231~price:null;sku:26780380131_13200,26780380132_8000,#
				//sb.append(valueText).append(SplitConstants.SPLIT_POUND);
			}
		}
		Set<String> keys = dateToUpdateCount.keySet();
		for(String timeKey : keys){
			//增加商品编辑记录的输出 rate_2012-02-12:0
			sb_rate.append(timeKey)
			.append(SplitConstants.SPLIT_COLON)
			.append(dateToUpdatePriceCount.get(timeKey))
			.append(SplitConstants.SPLIT_UNDER_BAR)
			.append(dateToUpdateCount.get(timeKey))
			.append(SplitConstants.SPLIT_POUND);
		}
		output.collect(key, new Text("count@"+sb_rate.toString()));
		
		//此处加判断是防止处理以rate开头的数据时出现多余的空数据
		String editPriceResult = sb_date.toString();
		if(StringUtils.isNotEmpty(editPriceResult)){
			//输出时间相关的信息
			//截取后格式：1_10  代表周一的10点
			//time@1_10#7_23#
			output.collect(key, new Text("time@"+sb_date.toString()));
			//detail信息不输出
			//格式如：
			//detail@1338607656231~price:null;sku:26780380131_13200,26780380132_8000,#
			//output.collect(key, new Text("detail@"+editPriceResult));
		}
		
	}

	/**
	 * 解析数据中的编辑时间
	 * 传入的数据源格式如：1338607656231~price:null;sku:26780380131_13200,26780380132_8000,
	 * 截取后格式：1_10  代表周一的10点
	 * 1_10#7_23#
	 * @param sb_date
	 * @param valueText
	 */
	private void addDateTimeMark(StringBuilder sb_date, String valueText) {
		if(StringUtils.isEmpty(valueText)){
			return;
		}
		String[] datas = valueText.split(SplitConstants.SPLIT_WAVY);
		if(datas.length <2){
			return;
		}
		String longTime = datas[0];
		try{
			Calendar calendar = Calendar.getInstance();
			Date d = new Date(Long.parseLong(longTime));
			calendar.setTime(d);
			//获得小时
			int hour = calendar.get(Calendar.HOUR_OF_DAY);
			//获得星期的时间，1为周日 2，3，4，5，6，7（周六），等会全部减1吧
			int dayOfWeek = calendar.get(Calendar.DAY_OF_WEEK);
			
			int covertDayOfWeek = dayOfWeek-1;
			
			if(covertDayOfWeek==0){
				//将周日的时间设置为7
				sb_date.append(7).append(SplitConstants.SPLIT_UNDER_BAR).append(hour);
			}else{
				sb_date.append(covertDayOfWeek).append(SplitConstants.SPLIT_UNDER_BAR).append(hour);
			}
			sb_date.append(SplitConstants.SPLIT_POUND);
		}catch(Exception e){
			
		}
	}

	/**
	 * 对商品编辑记录和编辑价格的次数进行记录
	 * @param valueText
	 * @param dateToUpdateCount
	 * @param dateToUpdatePriceCount
	 */
	private void recordUpdateCount(String valueText,
			Map<String, Integer> dateToUpdateCount,
			Map<String, Integer> dateToUpdatePriceCount) {
		//这里需要补注释，下次进来必晕
		String[] datas = valueText.split(SplitConstants.SPLIT_COLON);
		int countEditPrice = Integer.parseInt(datas[1]);
		String[] rate_time = datas[0].split(SplitConstants.SPLIT_UNDER_BAR);
		if(dateToUpdateCount.get(rate_time[1])==null){
			dateToUpdateCount.put(rate_time[1], 1);
			dateToUpdatePriceCount.put(rate_time[1], countEditPrice);
		}else{
			dateToUpdateCount.put(rate_time[1], dateToUpdateCount.get(rate_time[1])+1);
			dateToUpdatePriceCount.put(rate_time[1], dateToUpdatePriceCount.get(rate_time[1])+countEditPrice);
		}
	}
	
	public static void main(String[] args) {
		Map<String,Integer> test = new HashMap<String,Integer>();
		System.out.println(test.get("abc"));
	}

}
